package xyz.hubery.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import net.openhft.affinity.AffinityThreadFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ThreadFactory;

@Configuration
@EnableConfigurationProperties(value = DisruptorProperties.class)
public class DisruptorAutoConfiguration {


    public DisruptorProperties disruptorProperties;

    public DisruptorAutoConfiguration(DisruptorProperties disruptorProperties) {
        this.disruptorProperties = disruptorProperties;
    }

    @Bean
    public EventFactory<OrderEvent> eventFactory() {
        return new EventFactory<OrderEvent>() {
            @Override
            public OrderEvent newInstance() {
                return new OrderEvent();
            }
        };
    }

    @Bean
    public ThreadFactory threadFactory(){
        return new AffinityThreadFactory("Match-Handler:") ;
    }

    /**
     * 无锁高效的等待策略
     * @return
     */
    @Bean
    public WaitStrategy waitStrategy(){
        return new YieldingWaitStrategy();
    }


    /**
     * 创建一个 RingBuffer
     * eventFactory: 事件工厂
     * threadFactory: 我们执行者(消费者)的线程该怎么创建
     * waitStrategy : 等待策略: 当我们 ringBuffer 没有数据时,我们怎么等待
     */
    @Bean
    public RingBuffer<OrderEvent> ringBuffer(EventFactory<OrderEvent> eventFactory, ThreadFactory threadFactory, WaitStrategy waitStrategy, EventHandler<OrderEvent>[] eventHandlers) {
        // 构建Disruptor
        ProducerType producerType = disruptorProperties.isMultiProducer() ? ProducerType.MULTI : ProducerType.SINGLE;
        final Disruptor<OrderEvent> disruptor =
                new Disruptor<OrderEvent>(eventFactory, disruptorProperties.getRingBufferSize(), threadFactory, producerType, waitStrategy);

        // 设置默认的异常处理器
        disruptor.setDefaultExceptionHandler(new DisruptorHandlerException());

        // 设置消费者---每个消费者对应一个交易对,有多少个交易对,就有多少个eventHandlers, 事件来了后, 多个 eventHandlers 是并发执行的
        disruptor.handleEventsWith(eventHandlers);

        disruptor.start() ;// 开始监听
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        // 优雅停机，关闭资源
        Runtime.getRuntime().addShutdownHook(new Thread(disruptor::shutdown,"DisruptorShutdownThread"));

        return ringBuffer;
    }

    /**
     * 创建 DisruptorTemplate
     * @param ringBuffer
     * @return
     */
    @Bean
    public DisruptorTemplate disruptorTemplate(RingBuffer<OrderEvent> ringBuffer){
        return new DisruptorTemplate(ringBuffer) ;
    }
}
